package ckR

//
//import (
//	"bytes"
//	"errors"
//	"fmt"
//	"db2s/tableEr"
//	"reflect"
//	"time"
//
//	ea "db2s/encryptionAlgorithm"
//	"db2s/global"
//	"db2s/scheduleTask"
//	mq "db2s/topic-mq"
//	"runtime"
//	"strings"
//	"sync"
//	"sync/atomic"
//)
//
//// partitionMap 根据指定的partition name输出相应的shardname和partitionid
//func (sp SchedulePlan2) partitionMap(partitionName string, object string) (partitionN, shardName, partitionId string) {
//	switch object {
//	case "source":
//		partitionN = partitionName
//		shardName = sp.Object.PartitionData.Shard.Source[partitionName]
//		partitionId = sp.Object.PartitionData.ShardId.Source[partitionName]
//	case "dest":
//		partitionN = sp.Object.PartitionData.SourceDestNameMap[partitionName]
//		shardName = sp.Object.PartitionData.Shard.Dest[partitionN]
//		partitionId = sp.Object.PartitionData.ShardId.Dest[partitionN]
//	}
//	return
//}
//
//// error info dispos column不一致
//func errorDispos(_ int64, plan *SchedulePlan2, err string) (errstr string) {
//	if strings.EqualFold(err, "column length abnormal") { //表结构不一致
//		var srcColumn, dstColumn []string
//		var returnerror bytes.Buffer
//		for _, v := range plan.Object.TableColData.SColumnInfo {
//			srcColumn = append(srcColumn, v.ColumnName)
//		}
//		for _, v := range plan.Object.TableColData.DColumnInfo {
//			dstColumn = append(dstColumn, v.ColumnName)
//		}
//		aa := &ea.CheckSumTypeStruct{}
//		add, del := aa.Arrcmp(srcColumn, dstColumn)
//		for _, v := range add {
//			returnerror.WriteString(fmt.Sprintf("add column %v ,", v))
//		}
//		for _, v := range del {
//			returnerror.WriteString(fmt.Sprintf("drop column %v,", v))
//		}
//		errstr = returnerror.String()
//		returnerror.Reset()
//	} else {
//		errstr = err
//	}
//	return
//}
//func (sp SchedulePlan2) speedMonitor(done chan struct{}, wg *sync.WaitGroup) {
//	var (
//		lastSecondExecReadTime, lastSecondExecWriteTime   uint64
//		lastSecondExecReadCount, lastSecondExecWriteCount int64
//		lastSecondReadSum, lastSecondWriteSum             int64
//		speedRestricted                                   = "N"
//		speedRestriction                                  string
//		readRate, writeRate                               uint64
//	)
//	ll := time.NewTicker(time.Duration(1) * time.Second)
//	for {
//		select {
//		case <-ll.C:
//			if *sp.TableSpeedLimit.ReadSum <= 0 {
//				continue
//			}
//			var speedlimitStatus bool
//			//获取读取端和写入端的差异值
//			l := (*sp.TableSpeedLimit.ReadSum - *sp.TableSpeedLimit.WriteSum) * 100
//			p := l / (*sp.TableSpeedLimit.ReadSum)
//			//获取读取端每秒的速率
//			rq := *sp.TableSpeedLimit.ReadAvgTime - lastSecondExecReadTime
//			rc := *sp.TableSpeedLimit.ReadExecCount - lastSecondExecReadCount
//			if rq == 0 || rc == 0 {
//				readRate = 0
//			} else {
//				readRate = rq / uint64(rc)
//			}
//			//获取写入端每秒的速率
//			wq := *sp.TableSpeedLimit.WriteAvgTime - lastSecondExecWriteTime
//			wc := *sp.TableSpeedLimit.WriteExecCount - lastSecondExecWriteCount
//			if wq == 0 || wc == 0 {
//				writeRate = 0
//			} else {
//				writeRate = wq / uint64(wc)
//			}
//			//读写差异速率判断
//			if p >= 20 { //src data rate > dst data rate   ==> write rate low
//				speedRestriction = "RWD"
//				speedlimitStatus = true
//			} else if writeRate > 1000 { //写入效率太差
//				speedRestriction = "WR"
//				speedlimitStatus = true
//			} else if readRate > 1000 { //读取效率太差
//				speedRestriction = "RR"
//				speedlimitStatus = true
//			}
//			if speedlimitStatus {
//				speedRestricted = "Y"
//				atomic.SwapInt64(sp.TableSpeedLimit.Status, 1)
//				if sp.TableSpeedLimit.SpeedLimit.GetRefill("cap") >= 1000 {
//					sp.TableSpeedLimit.SpeedLimit.SetCapacity(sp.TableSpeedLimit.SpeedLimit.GetRefill("cap") / 1000)
//				}
//				if readRate > 1000 || writeRate > 1000 {
//					if atomic.LoadInt64(sp.ReadOptimizer.Thread) > 5 {
//						atomic.SwapInt64(sp.ReadOptimizer.Thread, atomic.LoadInt64(sp.ReadOptimizer.Thread)-1)
//					}
//					if atomic.LoadInt64(sp.writeOptimizer.Thread) > 1 {
//						atomic.SwapInt64(sp.writeOptimizer.Thread, atomic.LoadInt64(sp.writeOptimizer.Thread)-1)
//					}
//				}
//			} else {
//				speedRestricted = "N"
//				atomic.SwapInt64(sp.TableSpeedLimit.Status, 0)
//				if sp.TableSpeedLimit.SpeedLimit.GetRefill("cap") <= 100000000 {
//					sp.TableSpeedLimit.SpeedLimit.SetCapacity(sp.TableSpeedLimit.SpeedLimit.GetRefill("cap") * 10)
//				}
//				if readRate < 501 && rc > 0 && writeRate < 501 && wc > 0 {
//					atomic.SwapInt64(sp.ReadOptimizer.Thread, atomic.LoadInt64(sp.ReadOptimizer.Thread)+1)
//					atomic.SwapInt64(sp.writeOptimizer.Thread, atomic.LoadInt64(sp.writeOptimizer.Thread)+1)
//				}
//			}
//			if GlobalPConfigs.result.Teletypewriter == "bar" {
//				if PlanContext.TaskBarSubsStatusGet(sp.subTaskInfo) != -2 {
//					PlanContext.SpeedTaskBarAccumulate(mq.SpeedLimitMonitor{
//						Schema: sp.Object.Schema,
//						Table:  sp.Object.Table,
//						SS:     speedRestricted,
//						SR:     speedRestriction,
//						RS:     fmt.Sprintf("%v", *sp.TableSpeedLimit.ReadSum-lastSecondReadSum),
//						WS:     fmt.Sprintf("%v", *sp.TableSpeedLimit.WriteSum-lastSecondWriteSum),
//						RWD:    fmt.Sprintf("%v", p),
//						RR:     fmt.Sprintf("%v", readRate),
//						WR:     fmt.Sprintf("%v", writeRate),
//						RP:     fmt.Sprintf("%v", *sp.TableSpeedLimit.ReadParallel),
//						WP:     fmt.Sprintf("%v", *sp.TableSpeedLimit.WriteParallel),
//						BS:     fmt.Sprintf("%v", sp.TableSpeedLimit.SpeedLimit.GetRefill("cap")),
//						CGL:    fmt.Sprintf("%v", atomic.LoadInt64(sp.ReadOptimizer.Thread)),
//						RBS:    fmt.Sprintf("%v", sp.ReadOptimizer.ChunkSum),
//					})
//				}
//			}
//			lastSecondReadSum = *sp.TableSpeedLimit.ReadSum
//			lastSecondWriteSum = *sp.TableSpeedLimit.WriteSum
//			lastSecondExecReadTime = *sp.TableSpeedLimit.ReadAvgTime
//			lastSecondExecWriteTime = *sp.TableSpeedLimit.WriteAvgTime
//			lastSecondExecReadCount = *sp.TableSpeedLimit.ReadExecCount
//			lastSecondExecWriteCount = *sp.TableSpeedLimit.WriteExecCount
//		case <-done:
//			wg.Done()
//			return
//		}
//	}
//}
//
//// startWhereInit whereadd功能  根据库、表、索引列信息进行筛选，筛选出whereadd的sql语句和选择前导索引列
//func startWhereInit(schema, table string, indexColumn []string) (s, i string) {
//	if v, ok := GlobalPConfigs.schema.WhereAdd[fmt.Sprintf("%s%s%s", schema, global.SchemaTableSplit, table)]; ok && v != "" {
//		var idxcColumn string
//		var whereAdd = make(map[string]int)
//		for _, v1 := range strings.Split(strings.ToLower(v), " and ") {
//			if len(strings.Split(strings.TrimSpace(v1), " ")) > 0 {
//				whereAdd[strings.Split(strings.TrimSpace(v1), " ")[0]]++
//			}
//		}
//		if len(indexColumn) > 1 {
//			var ff []string
//			for _, v1 := range indexColumn {
//				if _, ok1 := whereAdd[strings.ToLower(v1)]; !ok1 {
//					ff = append(ff, v1)
//				}
//			}
//			if len(ff) == 0 {
//				return v, ""
//			}
//			idxcColumn = ff[0]
//		} else {
//			idxcColumn = indexColumn[0]
//		}
//		s, i = v, strings.ToUpper(idxcColumn)
//	} else {
//		i = indexColumn[0]
//	}
//	return
//}
//
//// srcMissIndexCount
//func srcMissIndexCount(wg *sync.WaitGroup, partitionName1 string, plan *SchedulePlan2, whereSql string, logseq int64) {
//	var (
//		event    = "[srcMissIndexCount]"
//		tableSum int64
//		vlog     string
//	)
//	wg.Add(1)
//	defer func() {
//		wg.Done()
//	}()
//	pc, _, _, _ := runtime.Caller(1)
//	callingFunc := runtime.FuncForPC(pc)
//	callFile, callLine := callingFunc.FileLine(pc)
//	callFunc := callingFunc.Name()
//	callFuncInfo := fmt.Sprintf("Call functions %v Call File %v:%v", callFunc, callFile, callLine)
//	var ShardName, PartitionId string
//	if strings.EqualFold(plan.Object.PartitionData.Logo.Source, "single") {
//		ShardName = "single"
//	} else {
//		_, ShardName, PartitionId = plan.partitionMap(partitionName1, "source")
//	}
//	if t3, err := tableEr.Sum(tableEr.TableSum{DBType: GlobalPConfigs.dSns.SrcDBName}).Count(global.TableSumInput{
//		TableInfo: global.TableInfo{
//			Schema:        plan.Object.Schema,
//			Table:         plan.Object.Table,
//			PartitionName: partitionName1,
//			PartitionId:   PartitionId,
//			ShardName:     ShardName,
//		},
//		SqlFilter: global.SqlFilter{
//			WhereSql: whereSql,
//		},
//		Db:     GlobalPConfigs.SDB,
//		LogSeq: logseq,
//	}); err != nil {
//		vlog = fmt.Sprintf("(%d) %v %v Failed to obtain table data from source database count. error info is {%v}", logseq, callFuncInfo, event, err)
//		global.Wlog.Error(vlog)
//		PlanContext.TaskBarSubsErrorSwap(plan.subTaskInfo, errorDispos(logseq, plan, fmt.Sprintf("%v", err)))
//		return
//	} else {
//		tableSum = int64(t3)
//	}
//	atomic.AddInt64(plan.rows.CheckRows.SrcCheckCount, tableSum)
//	atomic.AddInt64(plan.subTaskInfo.RecordCount, tableSum)
//	atomic.AddInt64(plan.TPod.Rows.RowsSum, tableSum)
//	vlog = fmt.Sprintf("(%d) %v %v Source database count obtains table data successfully. result is {%v}", logseq, callFuncInfo, event, tableSum)
//	global.Wlog.Debug(vlog)
//}
//
//// 出现错误集体处理
//func (sp SchedulePlan2) errorAction(message mq.Message, logseq int64, err any) {
//	taskObject, ok := message.Properties["taskObject"]
//	if ok {
//		if taskObject != nil {
//			atomic.SwapInt64(taskObject.(scheduleTask.TaskDistribution).ErrEnd, -1)
//		}
//	}
//	atomic.AddInt64(sp.Status.ErrEnd, -1)
//	PlanContext.TaskBarSubsErrorSwap(sp.subTaskInfo, errorDispos(logseq, &sp, fmt.Sprintf("%v", err)))
//	if PlanContext.TaskBarSubsStatusGet(sp.subTaskInfo) != -2 {
//		PlanContext.TaskBarSubsStatusSwap(sp.subTaskInfo, "error")
//	}
//	PlanContext.TPodTableStatusSwap(sp.TPod.Sync.SyncStatus, "error")
//}
//
//// whereSqlFilter sql 过滤
//func (sp *SchedulePlan2) whereSqlFilter(wg *sync.WaitGroup, message mq.Message, v string) (sqlWhere chan any) {
//	var (
//		topic     = QueryTableData
//		tableName = string(message.Body)
//		logseq    = message.Properties["logSeq"].(int64)
//		//insertPrefix string
//		//err          error
//	)
//	go srcMissIndexCount(wg, "single", sp, v, logseq)
//	//if insertPrefix, err = sps.ifTableLobColumn("single"); err != nil {
//	//	sps.errorAction(message, logseq, err)
//	//	return
//	//}
//	tableSql, err := sp.QuerySql(global.StartPart{
//		LogSeq: logseq,
//		TableInfo: global.TableInfo{
//			Schema: sp.Object.Schema,
//			Table:  sp.Object.Table,
//		},
//		SelectColumn: global.SelectColumnInput{
//			Schema:      sp.Object.Schema,
//			Table:       sp.Object.Table,
//			IndexColumn: sp.ReadOptimizer.IndexColumn,
//			TableColumn: sp.Object.TableColData.SColumnInfo,
//			IndexName:   sp.ReadOptimizer.IndexName,
//		},
//		WhereGenerate: global.WhereGenerateInput{
//			Schema:      sp.Object.Schema,
//			Table:       sp.Object.Table,
//			TableColumn: sp.Object.TableColData.SColumnInfo,
//			SqlFilter:   global.SqlFilter{WhereSql: v},
//		},
//	}, "single")
//	if err != nil {
//		return
//	}
//	if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"logSeq": logseq, "plan": message.Properties["plan"],
//		"taskObject": message.Properties["taskObject"], "label": message.Properties["label"],
//		"strsql": tableSql, "err": sp.Status.ErrEnd, "sourcePartitionName": "single",
//		//"insertPrefix": insertPrefix,
//		"subTask": sp.subTaskInfo, "topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}) {
//		return
//	}
//	if atomic.LoadInt64(sp.Status.ErrEnd) > -1 {
//		atomic.AddInt64(sp.mqProductCustomerMonitor.MqLevelProductSend, 1)
//		atomic.AddInt64(sp.mqProductCustomerMonitor.MqStartTiming, 1)
//		atomic.AddInt64(sp.mqProductCustomerMonitor.MqLevelCustomerReceive, 1)
//		atomic.AddInt64(sp.mqProductCustomerMonitor.MqProductSeq, 1)
//	}
//	return
//}
//func (sp SchedulePlan2) ifTableLobColumn(sourcePartitionName string) (string, error) {
//	var (
//		lobSwitch    = false
//		insertColumn []string
//		insertPrefix any
//		err          error
//	)
//	for _, v := range sp.Object.TableColData.DColumnInfo {
//		insertColumn = append(insertColumn, v.ColumnName)
//		dateType := strings.ToLower(v.DataType)
//		if strings.Contains(dateType, "lob") {
//			lobSwitch = true
//		}
//	}
//	var s1 []string
//	if lobSwitch {
//		for i := 0; i < len(insertColumn); i++ {
//			s1 = append(s1, "?")
//		}
//	}
//	partitionName, shardName, partitionId := sp.partitionMap(sourcePartitionName, "dest")
//	generalSqlInput := global.PrefixInput{
//		LogSeq: sp.Status.LogSeq,
//		Insert: global.Insert{
//			Prefix: global.InsertPrefix{
//				TableInfo: global.TableInfo{
//					Schema:        sp.Object.MSchema,
//					Table:         sp.Object.MTable,
//					PartitionName: partitionName,
//					ShardName:     shardName,
//					PartitionId:   partitionId,
//				},
//				PlaceholderValue: s1,
//				InsertColumn:     insertColumn,
//				LobSwitch:        lobSwitch,
//			},
//		},
//	}
//	if insertPrefix, err = tableEr.Sql(tableEr.TableSql{DBType: GlobalPConfigs.dSns.DestDBName}).Prefix(generalSqlInput); err != nil {
//		return "", err
//	}
//	return insertPrefix.(string), nil
//}
//func (sp *SchedulePlan2) IsSegmentSendMessage(wg *sync.WaitGroup, message mq.Message) {
//	var (
//		topic             = IndexConcurrentQuery
//		tableName         = string(message.Body)
//		taskObject1       = scheduleTask.TaskDistribution{}
//		logseq            = message.Properties["logSeq"].(int64)
//		errEnd      int64 = 0
//	)
//
//	taskObject1 = scheduleTask.TaskDistribution{
//		LogSeq:     logseq,
//		DataSource: sp.ReadOptimizer.ChunkStartSource,
//		TableObject: scheduleTask.TableObject{
//			Schema:  sp.Object.Schema,
//			Table:   sp.Object.Table,
//			MSchema: sp.Object.MSchema,
//			MTable:  sp.Object.MTable,
//		},
//		Db: scheduleTask.ConnDb{
//			SourceObject: GlobalPConfigs.dSns.SrcDBName,
//			DestObject:   GlobalPConfigs.dSns.DestDBName,
//			Sdb:          GlobalPConfigs.SDB,
//			Ddb:          GlobalPConfigs.DDB,
//		},
//		TableInfo: scheduleTask.TableInfo{
//			TableColumn: sp.Object.TableColData.SColumnInfo,
//			IndexColumn: sp.ReadOptimizer.IndexColumn,
//			IndexName:   sp.ReadOptimizer.IndexName,
//		},
//		StaticP: scheduleTask.StaticParameter{
//			CpLength:    GlobalPConfigs.rules.QueueSize,
//			MulFactor:   sp.ReadOptimizer.Thread,
//			ChunkNumber: sp.ReadOptimizer.ChunkSum,
//		},
//		ErrEnd: &errEnd,
//	}
//	for _, partitionName1 := range sp.Object.PartitionData.Name.Source {
//		go srcMissIndexCount(wg, partitionName1, sp, "", logseq)
//		wg.Add(1)
//		go func(wg1 *sync.WaitGroup, partitionName string, taskObject scheduleTask.TaskDistribution) {
//			var labelChecklist = make(map[string][]string)
//			var insertPrefix string
//			var err error
//			defer func() {
//				if atomic.LoadInt64(sp.Status.ErrEnd) > -1 {
//					atomic.AddInt64(sp.mqProductCustomerMonitor.MqLevelProductSend, int64(len(labelChecklist)))
//					atomic.AddInt64(sp.mqProductCustomerMonitor.MqStartTiming, 1)
//				}
//				wg1.Done()
//			}()
//			taskObject.TableObject.PartitionName = partitionName
//			_, taskObject.TableObject.ShardName, taskObject.TableObject.PartitionId = sp.partitionMap(partitionName, "source")
//			var tableWhereAdd string
//			if w1, ok1 := GlobalPConfigs.schema.WhereAdd[getSchemaToTableMap(sp.Object.Schema, sp.Object.Table)]; ok1 {
//				tableWhereAdd = w1
//			}
//			if insertPrefix, err = sp.ifTableLobColumn(partitionName); err != nil {
//				sp.errorAction(message, logseq, err)
//				return
//			}
//			sqlWhere := taskObject.IsSegmentValue(tableWhereAdd, sp.ReadOptimizer.IndexColumn)
//			//send message
//			l := chanSendMsg{
//				Object:   "first",
//				SendData: sqlWhere,
//				SendMsgInfo: SendMsgInfo{
//					CheckMod:  getCheckMod(),
//					Topic:     topic,
//					TableName: tableName,
//				},
//				SendMsg:          sendMsg,
//				SpeedLimit:       sp.TableSpeedLimit.SpeedLimit,
//				SpeedLimitSwitch: true,
//				Status:           sp.TableSpeedLimit.Status,
//				FatherSon: FatherSon{
//					FatherTags: map[string]any{"logSeq": logseq, "plan": message.Properties["plan"],
//						"taskObject": taskObject,
//						"sendMsgKey": "segmentSql", "label": map[string]any{"no": []string{}},
//						"err": sp.Status.ErrEnd, "subTask": sp.subTaskInfo, "topic": string(topic),
//						"insertPrefix": insertPrefix,
//						"outFay":       GlobalPConfigs.result.Teletypewriter, "sourcePartitionName": partitionName},
//					FatherSonTags: labelChecklist,
//				},
//				SendMsgEnd: func() {
//				},
//			}
//			if !chanMonitorSendMsg(&l) {
//			}
//		}(wg, partitionName1, taskObject1)
//	}
//}
//
//// QueryTableSql 针对表的所有列的数据类型进行处理，将列类型转换成字符串，例如时间类型
////func (sps *SchedulePlan2) QueryTableSql(sqlWhere map[string]string, logThreadSeq int64) (strsql, strsqlCrc32 string) {
////	var (
////		vlog            string
////		err             error
////		buff, buffCrc32 bytes.Buffer
////		ssql, dsql      string
////		event           = "[general query table sql]"
////	)
////	vlog = fmt.Sprintf("(%d) %s Start processing the block data verification query sql of the verification table ...", logThreadSeq, event)
////	global.Wlog.Debug(vlog)
////	//查询该表的列名和列信息xz
////	if ssql, err = dbExec.TQuery(sps.idxc).GeneratingQuerySql(sqlWhere, logThreadSeq); err != nil {
////		return
////	} else if ssql != "" {
////		buff.WriteString(ssql)
////		buff.WriteString(string(SdriveSql))
////	}
////	if ssql1 := dbExec.TQuery(sps.idxc).GeneratingEncryptionQuerySql(ssql, GlobalPConfigs.rules.RowsR.Options, logThreadSeq); ssql1 != "" {
////		buffCrc32.WriteString(ssql1)
////		buffCrc32.WriteString(string(SdriveSql))
////	}
////
////	if dsql, err = dbExec.TQuery(sps.dstIdxc).GeneratingQuerySql(sqlWhere, logThreadSeq); err != nil {
////		return
////	} else if dsql != "" {
////		buff.WriteString(dsql)
////		buff.WriteString(string(DdriveSql))
////	}
////	if ssql1 := dbExec.TQuery(sps.dstIdxc).GeneratingEncryptionQuerySql(dsql, GlobalPConfigs.rules.RowsR.Options, logThreadSeq); ssql1 != "" {
////		buffCrc32.WriteString(ssql1)
////		buffCrc32.WriteString(string(DdriveSql))
////	}
////	vlog = fmt.Sprintf("(%d) %s The block data verification query sql processing of the verification table is completed. !!!", logThreadSeq, event)
////	global.Wlog.Debug(vlog)
////	strsql = buff.String()
////	buff.Reset()
////	strsqlCrc32 = buffCrc32.String()
////	buffCrc32.Reset()
////	return strsql, strsqlCrc32
////}
//
//func (sp SchedulePlan2) QuerySql(s global.StartPart, partitionName string) (strSql querySql, err error) {
//	var (
//		vlog  string
//		event = "[tmpSyncTableQuerySql]"
//	)
//	pc, _, _, _ := runtime.Caller(1)
//	callingFunc := runtime.FuncForPC(pc)
//	callFile, callLine := callingFunc.FileLine(pc)
//	callFunc := callingFunc.Name()
//	callFuncInfo := fmt.Sprintf("Call functions %v Call File %v:%v", callFunc, callFile, callLine)
//	vlog = fmt.Sprintf("(%v) %v %s sync data source query chunk where is %v", s.LogSeq, callFuncInfo, event, s)
//	global.Wlog.Debug(vlog)
//	//查询该表的列名和列信息xz
//	for k, v := range map[string]string{"source": GlobalPConfigs.dSns.SrcDBName, "dest": GlobalPConfigs.dSns.DestDBName} {
//		s.TableInfo.PartitionName = partitionName
//		_, s.TableInfo.ShardName, s.TableInfo.PartitionId = sp.partitionMap(partitionName, v)
//		s.SelectColumn.IndexName = sp.ReadOptimizer.IndexName
//		switch k {
//		case "source":
//			s.TableInfo.Schema = sp.Object.Schema
//			s.TableInfo.Table = sp.Object.Table
//			if s1, err1 := tableEr.RowsData(tableEr.TableRows{DBType: v}).Sql(s); err1 != nil {
//				err = err1
//				return
//			} else if len(s1) == 0 {
//				err = errors.New(fmt.Sprintf("general source query sql is empty"))
//				return
//			} else {
//				strSql.Source = s1
//			}
//		case "dest":
//			s.TableInfo.Schema = sp.Object.MSchema
//			s.TableInfo.Table = sp.Object.MTable
//			if s1, err1 := tableEr.RowsData(tableEr.TableRows{DBType: v}).Sql(s); err1 != nil {
//				err = err1
//				return
//			} else if len(s1) == 0 {
//				err = errors.New(fmt.Sprintf("general trigger query sql is empty"))
//				return
//			} else {
//				strSql.Dest = s1
//			}
//		}
//	}
//	vlog = fmt.Sprintf("(%v) %v %s sync data source query sql is {%v}", s.LogSeq, callFuncInfo, event, strSql)
//	global.Wlog.Debug(vlog)
//	return
//}
//
//// 根据返回数据进行行数统计和数据切割
//func (sp *SchedulePlan2) QueryTableData(sd querySql, partitionName string, logSeq int64) (sqld sqlData, err error) {
//	var (
//		vlog  string
//		event = "[rowsDataCountSplist]"
//		qSql  map[string]any
//	)
//	pc, _, _, _ := runtime.Caller(1)
//	callingFunc := runtime.FuncForPC(pc)
//	callFile, callLine := callingFunc.FileLine(pc)
//	callFunc := callingFunc.Name()
//	callFuncInfo := fmt.Sprintf("Call functions %v Call File %v:%v", callFunc, callFile, callLine)
//	vlog = fmt.Sprintf("(%d) %v %v start Query row data of table %s.%s", logSeq, callFuncInfo, event, sp.Object.Schema, sp.Object.Table)
//	global.Wlog.Debug(vlog)
//	if qSql, err = global.StructSubToMapAny(sd); err != nil {
//		return
//	}
//	for k, v := range map[string]string{
//		"source": GlobalPConfigs.dSns.SrcDBName,
//		"dest":   GlobalPConfigs.dSns.DestDBName,
//	} {
//		sqlStr := qSql[k]
//		l := global.RowsDataInput{
//			LogSeq: logSeq,
//			Sql:    fmt.Sprintf("%v", sqlStr),
//		}
//		switch k {
//		case "source":
//			l.DB = GlobalPConfigs.SDB
//		case "dest":
//			l.DB = GlobalPConfigs.DDB
//		}
//		_, l.TableInfo.ShardName, l.TableInfo.BackendTableName = sp.partitionMap(partitionName, k)
//		var stt any
//		if stt, err = tableEr.RowsData(tableEr.TableRows{DBType: v}).Data(l); err != nil {
//			vlog = fmt.Sprintf("(%d) %v %v The data query of table %s.%s fails. error is {%v}", logSeq, callFuncInfo, event, sp.Object.Schema, sp.Object.Table, err)
//			global.Wlog.Error(vlog)
//			return
//		} else {
//			var ssum int64
//			switch fmt.Sprintf("%v", reflect.TypeOf(stt.(global.ExecSqlResult).Result)) {
//			case "[]*string":
//				ssum = int64(len(stt.(global.ExecSqlResult).Result.([]*string)))
//			case "[][]*string":
//				ssum = int64(len(stt.(global.ExecSqlResult).Result.([][]*string)))
//			}
//			switch k {
//			case "source":
//				sqld.Source = stt.(global.ExecSqlResult).Result
//				sqld.Ssum = ssum
//				atomic.AddUint64(sp.TableSpeedLimit.ReadAvgTime, uint64(stt.(global.ExecSqlResult).Timeout))
//				atomic.AddInt64(sp.TableSpeedLimit.ReadExecCount, 1)
//				atomic.AddInt64(sp.TableSpeedLimit.ReadSum, ssum)
//				atomic.AddInt64(sp.TPod.Rows.SourceNumber, ssum)
//			case "dest":
//				sqld.Dest = stt.(global.ExecSqlResult).Result
//				sqld.Dsum = ssum
//				atomic.AddUint64(sp.TableSpeedLimit.WriteAvgTime, uint64(stt.(global.ExecSqlResult).Timeout))
//				atomic.AddInt64(sp.TableSpeedLimit.WriteExecCount, 1)
//				atomic.AddInt64(sp.TableSpeedLimit.WriteSum, ssum)
//				atomic.AddInt64(sp.TPod.Rows.DestNumber, ssum)
//			}
//		}
//	}
//	return
//}
//
//// Md5Rows 针对表的所有列的数据类型进行处理，将列类型转换成字符串，例如时间类型，并执行sql语句
//func (sp *SchedulePlan2) Md5Rows(stt, dtt any, logThreadSeq int64) (any, any, bool) {
//	var (
//		vlog  string
//		aa    = &ea.CheckSumTypeStruct{}
//		event string
//	)
//	global.Wlog.Info(fmt.Sprintf("(%d) %v Perform md5 check on the row data of table %s.%s.", logThreadSeq, event, sp.Object.Schema, sp.Object.Table))
//	if aa.CheckMd5(stt) != aa.CheckMd5(dtt) {
//		global.Wlog.Info(fmt.Sprintf("(%d) %v There is a discrepancy in the md5 of the row data of table %s.%s.", logThreadSeq, event, sp.Object.Schema, sp.Object.Table))
//		return stt, dtt, true
//	} else {
//		global.Wlog.Debug(fmt.Sprintf("(%d) %v The row data md5 of table %s.%s is the same.", logThreadSeq, event, sp.Object.Schema, sp.Object.Table))
//		global.DataSetFinalizer(stt)
//		global.DataSetFinalizer(dtt)
//		vlog = fmt.Sprintf("(%d) %v Clean up the row data memory of table %s.%s.", logThreadSeq, event, sp.Object.Schema, sp.Object.Table)
//		global.Wlog.Debug(vlog)
//	}
//	vlog = fmt.Sprintf("(%d) %v The md5 verification of the row data of table %s.%s is completed.", logThreadSeq, event, sp.Object.Schema, sp.Object.Table)
//	global.Wlog.Debug(vlog)
//	return nil, nil, false
//}
//
//// Arrcmp 针对表的所有列的数据类型进行处理，将列类型转换成字符串，例如时间类型，并执行sql语句
//func (sp *SchedulePlan2) Arrcmp(stt, dtt any, logThreadSeq int64) map[string]any {
//	var (
//		aa    = &ea.CheckSumTypeStruct{}
//		event string
//		A     = make(map[string]any)
//	)
//	global.Wlog.Info(fmt.Sprintf("(%d) [%s] Perform rows data arrcmp of table %s.%s.", logThreadSeq, event, sp.Object.Schema, sp.Object.Table))
//	add, del := aa.Arrcmp1(stt, dtt)
//	if del != nil {
//		A[string(Ldelete)] = del
//	}
//	if add != nil {
//		A[string(Linsert)] = add
//	}
//	return A
//}
//
//// 对比两个string中存在的差异
//func (sp *SchedulePlan2) stringArrcmp(s, d string, p string) (s1 string) {
//	//var (
//	//	m = make(map[string]string)
//	//)
//	//
//	//ff := func(s1, p1 string) (m1 map[int]string) {
//	//	m1 = make(map[int]string)
//	//	for k, v := range strings.Split(s1, p1) {
//	//		m1[k] = v
//	//	}
//	//	return
//	//}
//	//m1 := ff(d, p)
//	//var m2 []string
//	//for k, v := range ff(s, p) {
//	//	if v1, ok := sps.indexColumnSeq[fmt.Sprintf("%d", k+1)]; ok {
//	//		m2 = append(m2, fmt.Sprintf("\"%s\":\"%s\"", v1, v))
//	//	}
//	//	if v1, ok := m1[k]; ok {
//	//		if v != v1 {
//	//			m[sps.columnSeq[fmt.Sprintf("%d", k+1)]] = fmt.Sprintf("{\"src\":%s,\"dst\":%s}", v, v1)
//	//		}
//	//	}
//	//}
//	//var ss bytes.Buffer
//	//for k, v := range m {
//	//	ss.WriteString(fmt.Sprintf("%s:%s ", k, v))
//	//}
//	//s1 = fmt.Sprintf("(%d) [update] schema:%s table:%s index column:[{%s}] updateData: [%s]", sps.logThreadSeq, sps.schema, sps.table, strings.Join(m2, ","), ss.String())
//	//ss.Reset()
//	//atomic.AddInt64(terminalPods[getTableName(sps.schema, sps.table)].Rows.UpdateSum, 1)
//	return
//}
//
//// 对比两个string中存在的差异
//func (sp *SchedulePlan2) abnormalDataUpdateQuestionType(data map[string]string) (s1 *string) {
//	//var (
//	//	s, d string
//	//	p    = string(ColumnDataSplict)
//	//	m    = make(map[string]string)
//	//)
//	//if v, ok := data["add"]; ok {
//	//	s = v
//	//}
//	//if v, ok := data["del"]; ok {
//	//	d = v
//	//}
//	//stringSplist := func(s1, p1 string) (m1 map[int]string) {
//	//	m1 = make(map[int]string)
//	//	for k, v := range strings.Split(s1, p1) {
//	//		m1[k] = v
//	//	}
//	//	return
//	//}
//	//destValue := stringSplist(d, p)
//	//var indexValue []string
//	//for k, v := range stringSplist(s, p) {
//	//	if v1, ok := sps.indexColumnSeq[fmt.Sprintf("%d", k+1)]; ok {
//	//		indexValue = append(indexValue, fmt.Sprintf("\"%s\":\"%s\"", v1, v))
//	//	}
//	//	if v1, ok := destValue[k]; ok {
//	//		if v != v1 {
//	//			columnName := sps.columnSeq[fmt.Sprintf("%d", k+1)]
//	//			var columnType string
//	//			for _, v2 := range sps.tableColData.SColumnInfo {
//	//				if strings.EqualFold(v2["columnName"], columnName) {
//	//					columnType = dbExec.TQuery(sps.idxc).ColumnDataTypeJudge(columnName, 1)
//	//					switch columnType {
//	//					case "int":
//	//						atomic.AddInt64(sps.abnormalDataAnalyze.intTotal, 1)
//	//					case "string":
//	//						atomic.AddInt64(sps.abnormalDataAnalyze.stringTotal, 1)
//	//					case "time":
//	//						atomic.AddInt64(sps.abnormalDataAnalyze.timeTotal, 1)
//	//					}
//	//					break
//	//				}
//	//			}
//	//			m[columnName] = fmt.Sprintf("{\"issue Type\":%s, \"src\":%s,\"dst\":%s}", columnType, v, v1)
//	//			atomic.AddInt64(sps.abnormalDataAnalyze.abTotal, 1)
//	//		}
//	//	}
//	//}
//	//var ss bytes.Buffer
//	//for k, v := range m {
//	//	ss.WriteString(fmt.Sprintf("%s:%s ", k, v))
//	//}
//	//s2 := fmt.Sprintf("(%d) [update] schema:%s table:%s index column:[{%s}] updateData: [%s]", sps.logThreadSeq, sps.schema, sps.table, strings.Join(indexValue, ","), ss.String())
//	//ss.Reset()
//	//s1 = &s2
//	//atomic.AddInt64(terminalPods[getTableName(sps.schema, sps.table)].Rows.UpdateSum, 1)
//	return
//}
//
//func (sp *SchedulePlan2) abnormalDataInsertQuestionType(data map[string]string) (s1 *string) {
//	//var (
//	//	m2 []string
//	//	s  string
//	//)
//	//if v, ok := data["add"]; ok {
//	//	s = v
//	//}
//	//for k1, v2 := range strings.Split(s, string(ColumnDataSplict)) {
//	//	if v3, ok1 := sps.indexColumnSeq[fmt.Sprintf("%d", k1+1)]; ok1 {
//	//		m2 = append(m2, fmt.Sprintf("\"%s\":\"%s\"", v3, v2))
//	//		atomic.AddInt64(sps.abnormalDataAnalyze.missTotal, 1)
//	//		atomic.AddInt64(sps.abnormalDataAnalyze.abTotal, 1)
//	//	}
//	//}
//	//s2 := fmt.Sprintf("(%d) [insert] schema:%s table:%s \"issue Type\":%s index column:[{%s}] ", sps.logThreadSeq, sps.schema, sps.table, "missRows", strings.Join(m2, ","))
//	//s1 = &s2
//	return
//}
//func (sp *SchedulePlan2) abnormalDataDeleteQuestionType(data map[string]string) (s1 *string) {
//	//var (
//	//	m2 []string
//	//	d  string
//	//)
//	//if v, ok := data["del"]; ok {
//	//	d = v
//	//}
//	//for k1, v2 := range strings.Split(d, string(ColumnDataSplict)) {
//	//	if v3, ok1 := sps.indexColumnSeq[fmt.Sprintf("%d", k1+1)]; ok1 {
//	//		m2 = append(m2, fmt.Sprintf("\"%s\":\"%s\"", v3, v2))
//	//		atomic.AddInt64(sps.abnormalDataAnalyze.redundantTotal, 1)
//	//		atomic.AddInt64(sps.abnormalDataAnalyze.abTotal, 1)
//	//	}
//	//}
//	//s2 := fmt.Sprintf("(%d) [delete] schema:%s table:%s \"issue Type\":%s index column:[{%s}]", sps.logThreadSeq, sps.schema, sps.table, "moreRows", strings.Join(m2, ","))
//	//s1 = &s2
//	return
//}
//
//// 针对问题类型进行判断，包含类型：缺失数据、多余数据、字符差异、null值约束、时间类型、数值类型
//func (sp *SchedulePlan2) abnormalDataQuestionType(dataType string, data map[string]string) *string {
//	switch dataType {
//	case "insert":
//		return sp.abnormalDataInsertQuestionType(data)
//	case "delete":
//		return sp.abnormalDataDeleteQuestionType(data)
//	case "update":
//		return sp.abnormalDataUpdateQuestionType(data)
//	}
//	return nil
//}
//
//// 该函数针对一段差异数据进行分拣，按照insert、delete、update进行分类
//func (sp *SchedulePlan2) abnormalDataSorting(addM, delM map[string][]string) (ss []*string) {
//	for k, v := range addM {
//		var abType string
//		var dataV map[string]string
//		if v1, ok := delM[k]; ok { //update
//			abType = "update"
//			dataV = map[string]string{"add": strings.Join(v, ""), "del": strings.Join(v1, "")}
//			delete(delM, k)
//		} else { //insert
//			abType = "insert"
//			dataV = map[string]string{"add": strings.Join(v, "")}
//		}
//		ss = append(ss, sp.abnormalDataQuestionType(abType, dataV))
//	}
//	for _, v := range delM {
//		var abType = "delete"
//		var dataV = map[string]string{"del": strings.Join(v, "")}
//		ss = append(ss, sp.abnormalDataQuestionType(abType, dataV))
//	}
//	return
//}
//
//func (sp *SchedulePlan2) trxSumMerge(p [][]*string) (ll [][][]*string) {
//	trxNumber := int64(len(p))
//	if trxNumber > sp.writeOptimizer.FixSum {
//		var sum int64
//		var mm [][]*string
//		for _, v := range p {
//			sum++
//			if sum <= sp.writeOptimizer.FixSum {
//				mm = append(mm, v)
//			} else {
//				ll = append(ll, mm)
//				sum = 0
//				mm = [][]*string{v}
//			}
//		}
//		if len(mm) > 0 {
//			ll = append(ll, mm)
//		}
//		mm = nil
//	} else {
//		ll = append(ll, p)
//	}
//	return
//}
//
//func (sp *SchedulePlan2) multipleLevelsPointGeneralInsert(insertPrefix string, vv [][]*string, logSeq int64) (row any, err error) {
//	var (
//		vlog  string
//		event = "[multipleLevelsPointGeneralInsert]"
//	)
//	pc, _, _, _ := runtime.Caller(1)
//	callingFunc := runtime.FuncForPC(pc)
//	callFile, callLine := callingFunc.FileLine(pc)
//	callFunc := callingFunc.Name()
//	callFuncInfo := fmt.Sprintf("Call functions %v Call File %v:%v", callFunc, callFile, callLine)
//	if len(vv) == 0 {
//		return
//	}
//	generalSqlInput := global.PrefixInput{LogSeq: logSeq, Insert: global.Insert{
//		Value: global.InsertValue{
//			TableInfo: global.TableInfo{
//				Schema: sp.Object.Schema, Table: sp.Object.Table,
//			},
//			Prefix:             insertPrefix,
//			RowData:            vv,
//			ColData:            sp.Object.TableColData.DColumnInfo,
//			IgnoreColumnLength: GlobalPConfigs.rules.Sync.IgnoreColumnLength,
//			Esa:                sp.Object.Esa,
//		},
//	}}
//	atomic.AddInt64(sp.TPod.Rows.InsertSum, 1)
//	if row, err = tableEr.Sql(tableEr.TableSql{DBType: GlobalPConfigs.dSns.DestDBName}).Insert(generalSqlInput); err != nil || row == nil {
//		vlog = fmt.Sprintf("(%v) %v %v Failed to generate insert statement for table %v.%v. error is {%v}", logSeq, callFuncInfo, event, sp.Object.Schema, sp.Object.Table, err)
//		global.Wlog.Error(vlog)
//		atomic.SwapInt64(sp.Status.ErrEnd, -1)
//		return nil, err
//	}
//	return
//}
//func (sp *SchedulePlan2) multipleLevelsPointGeneralDelete(vv [][]*string, logSeq int64) (row any, err error) {
//	var (
//		vlog  string
//		event = "[multipleLevelsPointGeneralDelete]"
//	)
//	pc, _, _, _ := runtime.Caller(1)
//	callingFunc := runtime.FuncForPC(pc)
//	callFile, callLine := callingFunc.FileLine(pc)
//	callFunc := callingFunc.Name()
//	callFuncInfo := fmt.Sprintf("Call functions %v Call File %v:%v", callFunc, callFile, callLine)
//	if len(vv) == 0 {
//		return
//	}
//	generalSqlInput := global.PrefixInput{LogSeq: logSeq, Delete: global.Delete{
//		Value: global.DeleteValue{
//			TableInfo: global.TableInfo{
//				Schema: sp.Object.MSchema,
//				Table:  sp.Object.MTable,
//			},
//			RowData:        vv,
//			IndexType:      sp.ReadOptimizer.IndexColumnType,
//			IndexName:      sp.ReadOptimizer.IndexName,
//			IndexColumnSeq: sp.ReadOptimizer.IndexColumnSeq,
//			ColData:        sp.Object.TableColData.SColumnInfo,
//			Esa:            sp.Object.Esa,
//		},
//	}}
//	atomic.AddInt64(sp.TPod.Rows.DeleteSum, 1)
//	if row, err = tableEr.Sql(tableEr.TableSql{DBType: GlobalPConfigs.dSns.DestDBName}).Delete(generalSqlInput); err != nil || row == nil {
//		vlog = fmt.Sprintf("(%v) %v %v Failed to generate insert statement for table %v.%v. error is {%v}", logSeq, callFuncInfo, event, sp.Object.Schema, sp.Object.Table, err)
//		global.Wlog.Error(vlog)
//		atomic.SwapInt64(sp.Status.ErrEnd, -1)
//		return
//	}
//	return
//}
//func (sp *SchedulePlan2) multipleLevelsPointGeneralFixSql(insertPrefix string, ll [][][]*string, t Logo, logSeq int64) (fix []any, err error) {
//	var (
//		insertChan = make(chan any, GlobalPConfigs.rules.QueueSize)
//		deleteChan = make(chan any, GlobalPConfigs.rules.QueueSize)
//		vlog       string
//		event      = "[multipleLevelsPointGeneralFixSql]"
//	)
//	go func() {
//		defer func() {
//			close(insertChan)
//			close(deleteChan)
//		}()
//		var wg sync.WaitGroup
//		for _, v := range ll {
//			if atomic.LoadInt64(sp.Status.ErrEnd) < 0 {
//				return
//			}
//			wg.Add(1)
//			go func(vv [][]*string) {
//				defer func() {
//					global.DataSetFinalizer(vv)
//					wg.Done()
//				}()
//				switch t {
//				case Linsert:
//					var i any
//					if i, err = sp.multipleLevelsPointGeneralInsert(insertPrefix, vv, logSeq); err != nil {
//						return
//					} else {
//						if i != nil {
//							insertChan <- i
//						}
//					}
//				case Ldelete:
//					var i any
//					if i, err = sp.multipleLevelsPointGeneralDelete(vv, logSeq); err != nil {
//						return
//					} else {
//						if i != nil {
//							deleteChan <- i
//						}
//					}
//				}
//			}(v)
//		}
//		wg.Wait()
//	}()
//	switch t {
//	case Linsert:
//		for result := range insertChan {
//			fix = append(fix, result)
//		}
//	case Ldelete:
//		for result := range deleteChan {
//			fix = append(fix, result)
//		}
//	}
//	vlog = fmt.Sprintf("(%v) %v The insert statement of table %v.%v is generated. Generate repair transactions according to the number of fixes.", logSeq, event, sp.Object.Schema, sp.Object.Table)
//	global.Wlog.Debug(vlog)
//	return fix, nil
//}
//
//// AbnormalDataDiff 差异数据的二次校验，并生成修复语句
//func (sp *SchedulePlan2) AbnormalDataDiff(diffQueryData map[string]any, insertPrefix string, logSeq int64) (fix FixSql, err error) {
//	var (
//		vlog  string
//		event = "[AbnormalDataDiff]"
//	)
//	vlog = fmt.Sprintf("(%d) Check table %s.%s to start differential data processing and generate repair statements ...", logSeq, sp.Object.Schema, sp.Object.Table)
//	global.Wlog.Info(vlog)
//	if del, ok := diffQueryData[string(Ldelete)]; ok && del != nil {
//		global.Wlog.Debug(fmt.Sprintf("(%d) Check table %s.%s abnormal data delete is %v ", logSeq, sp.Object.Schema, sp.Object.Table))
//		vlog = fmt.Sprintf("(%d) %v Start generating table %s.%s delete or insert statement", logSeq, event, sp.Object.Schema, sp.Object.Table)
//		global.Wlog.Debug(vlog)
//		switch fmt.Sprintf("%v", reflect.TypeOf(del)) {
//		case "[][]*string":
//			ll := sp.trxSumMerge(del.([][]*string))
//			if fix.Delete, err = sp.multipleLevelsPointGeneralFixSql(insertPrefix, ll, Ldelete, logSeq); err != nil {
//				return
//			}
//		}
//		vlog = fmt.Sprintf("(%d) %v The repair statement for table %s.%s is generated.", logSeq, event, sp.Object.Schema, sp.Object.Table)
//		global.Wlog.Debug(vlog)
//	}
//	if add, ok := diffQueryData[string(Linsert)]; ok && add != nil {
//		switch fmt.Sprintf("%v", reflect.TypeOf(add)) {
//		case "[][]*string":
//			ll := sp.trxSumMerge(add.([][]*string))
//			global.Wlog.Debug(fmt.Sprintf("(%d) Check table %s.%s abnormal data insert is %v ", logSeq, sp.Object.Schema, sp.Object.Table))
//			if fix.Insert, err = sp.multipleLevelsPointGeneralFixSql(insertPrefix, ll, Linsert, logSeq); err != nil {
//				return
//			}
//		}
//	}
//	vlog = fmt.Sprintf("(%d) Check table %s.%s to complete differential data processing and generate repair statements. !!!", logSeq, sp.Object.Schema, sp.Object.Table)
//	global.Wlog.Info(vlog)
//	return
//}
//
//func (sp SchedulePlan2) writeFixSql(lobSwitch bool, prefix string, s FixSql, sourcePartitionName string, logseq int64) (err error) {
//	var (
//		vlog  string
//		event = "[writeFixSql]"
//	)
//	pc, _, _, _ := runtime.Caller(1)
//	callingFunc := runtime.FuncForPC(pc)
//	callFile, callLine := callingFunc.FileLine(pc)
//	callFunc := callingFunc.Name()
//	callFuncInfo := fmt.Sprintf("Call functions %v Call File %v:%v", callFunc, callFile, callLine)
//	vlog = fmt.Sprintf("(%v) %v %v Table %v.%v starts executing the insert statement.", logseq, callFuncInfo, event, sp.Object.Schema, sp.Object.Table)
//	global.Wlog.Debug(vlog)
//	// 切片为空
//	var partition, ShardName, PartitionId string
//	if destpartition, ok := sp.Object.PartitionData.SourceDestNameMap[sourcePartitionName]; ok {
//		partition = destpartition
//	}
//	if l, ok := sp.Object.PartitionData.Shard.Dest[partition]; ok {
//		ShardName = l
//	} else {
//		ShardName = "single"
//	}
//	if l, ok := sp.Object.PartitionData.ShardId.Dest[partition]; ok {
//		PartitionId = l
//	}
//	rowsActive := global.RowsActiveInput{
//		LogSeq: logseq,
//		DB:     GlobalPConfigs.DDB,
//		TableInfo: global.TableInfo{
//			Schema:        sp.Object.Schema,
//			Table:         sp.Object.Table,
//			LobSwitch:     lobSwitch,
//			PartitionName: partition,
//			ShardName:     ShardName,
//			PartitionId:   PartitionId,
//		},
//		Optimizer: global.WriteOptimizer{
//			Active:    GlobalPConfigs.repair.DataFix,
//			FilePoint: sp.repair.FixFileDmlF,
//			SqlMode:   GlobalPConfigs.rules.Sync.SqlMode,
//			SqlLogBin: GlobalPConfigs.rules.Sync.SqlLogBin,
//		},
//		SqlPrefix: prefix,
//	}
//	if len(s.Delete) > 0 {
//		rowsActive.ApplySqlGather = s.Delete
//		if _, err = tableEr.RActive(tableEr.TableRows{DBType: GlobalPConfigs.dSns.DestDBName}).Delete(rowsActive); err != nil {
//			vlog = fmt.Sprintf("(%v) %v %v The insert statement of table %v.%v executes incorrectly!!! error is {%v}", logseq, callFuncInfo, event, sp.Object.Schema, sp.Object.Table, err)
//			global.Wlog.Error(vlog)
//			return err
//		}
//	}
//	if len(s.Insert) > 0 {
//		rowsActive.ApplySqlGather = s.Insert
//		if _, err = tableEr.RActive(tableEr.TableRows{DBType: GlobalPConfigs.dSns.DestDBName}).Insert(rowsActive); err != nil {
//			vlog = fmt.Sprintf("(%v) %v %v The insert statement of table %v.%v executes incorrectly!!! error is {%v}", logseq, callFuncInfo, event, sp.Object.Schema, sp.Object.Table, err)
//			global.Wlog.Error(vlog)
//			return err
//		}
//	}
//	vlog = fmt.Sprintf("(%v) %v %v The insert statement of table %v.%v is completed.", logseq, callFuncInfo, event, sp.Object.Schema, sp.Object.Table)
//	global.Wlog.Debug(vlog)
//	return
//}
